package com.whale.api;

import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.parquet.ParquetWriterFactory;
import org.apache.flink.formats.parquet.avro.ParquetAvroWriters;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy;

import java.util.Map;

/**
 * 列格式FileSink.forRowFormat
 */
public class _08_FIleSInk_Bulk {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(2);

        DataStreamSource<EventLog> streamSource = env.addSource(new MySourceFunction());

        // 输出到文件
        streamSource.map(bean -> Tuple5.of(bean.getEventId(), bean.getGuid(), bean.getEventInfo(), bean.getSessionId(), bean.getTimeStamp())).returns(new TypeHint<Tuple5<String, Long, Map<String, String>, String, Long>>() {
        });


        /**
         * 应用  StreamFileSink 算子，来将数据输出到  文件系统
         */

        /**
         * 2. 输出为 列格式
         *
         * 要构造一个列模式的 FileSink，需要一个ParquetAvroWriterFactory
         * 而获得ParquetAvroWriterFactory的方式是，利用一个工具类： ParquetAvroWriters
         * 这个工具类提供了3种方法，来为用户构造一个ParquetAvroWriterFactory
         *
         * ## 方法1：
         * writerFactory = ParquetAvroWriters.forGenericRecord(schema)
         *
         * ## 方法2：
         * writerFactory = ParquetAvroWriters.forSpecificRecord(AvroEventLogBean.class)
         *
         * ## 方法3：
         * writerFactory = ParquetAvroWriters.forReflectRecord(EventLog.class);
         *
         * 一句话：  3种方式需要传入的参数类型不同
         * 1. 需要传入schema对象
         * 2. 需要传入一种特定的JavaBean类class
         * 3. 需要传入一个普通的JavaBean类class
         *
         * 传入这些参数，有何用？
         * 这个工具 ParquetAvroWriters. 需要你提供你的数据模式schema（你要生成的parquet文件中数据模式schema）
         *
         * 上述的3种参数，都能让这个工具明白你所指定的数据模式（schema）
         *
         * 1. 传入Schema类对象，它本身就是parquet框架中用来表达数据模式的内生对象
         *
         * 2. 传入特定JavaBean类class，它就能通过调用传入的类上的特定方法，来获得Schema对象
         * （这种特定JavaBean类，不用开发人员自己去写，而是用avro的schema描述文件+代码生产插件，来自动生成   mvn clean+compile）
         *
         * 3. 传入普通JavaBean,然后工具可以自己通过反射手段来获取用户的普通JavaBean中的包名、类名、字段名、字段类型等信息，来翻译成一个符合Avro要求的Schema
         *
         *
         */
        // 方式 1：（只是演示，正常不这么用） 手动构造schema，来生成ParquetAvroWriter工厂
//        Schema schema = Schema.createRecord("id", "用户id", "cn.doitedu.User", true);
//        ParquetWriterFactory<GenericRecord> writerFactory0 = ParquetAvroWriters.forGenericRecord(schema);

        // 方式2：为了生成Schema，需要构建一个JSON，它要符合Avro格式，是我们javaBean的DDL，mvn clean+compile即可生成对应的AvroEventLogBean
//        ParquetWriterFactory<AvroEventLogBean> writerFactory1 = ParquetAvroWriters.forSpecificRecord(AvroEventLogBean.class);
//        streamSource
//                .map(eventLogBean -> {
//                    HashMap<CharSequence, CharSequence> eventInfo = new HashMap<>();
//                    // 这里需要转一下，不然new AvroEventLogBean()最后一个参数传不过去，原理：A是B的父亲，但是List<A>不是List<B>的父亲
//                    for (Map.Entry<String, String> entry : eventLogBean.getEventInfo().entrySet()) {
//                        eventInfo.put(entry.getKey(), entry.getValue());
//                    }
//                    return new AvroEventLogBean(eventLogBean.getGuid(), eventLogBean.getSessionId(), eventLogBean.getEventId(), eventLogBean.getTimeStamp(), eventInfo);
//                }).returns(AvroEventLogBean.class).sinkTo(parquetSink1);

        // 方式 3：（推荐使用）直接使用普通的javaBean即可。原理就是利用反射，把Bean生成一个这个Bean对应的DDL（只不过是JSON格式）
        ParquetWriterFactory<EventLog> writerFactory2 = ParquetAvroWriters.forReflectRecord(EventLog.class);// 该方法，传入一个普通的JavaBean类，就可以自动通过反射来生成 Schema

        FileSink<EventLog> parquetSink2 = FileSink
                .forBulkFormat(new Path("d:/filesink"), writerFactory2)
                .withBucketAssigner(new DateTimeBucketAssigner<EventLog>())
                .withBucketCheckInterval(5)
                .withRollingPolicy(OnCheckpointRollingPolicy.build())  // bulk模式下的文件滚动策略，只有一种： 当 checkpoint发生时，进行文件滚动
                .withOutputFileConfig(OutputFileConfig.builder().withPartSuffix(".parquet").withPartPrefix("doitedu").build())
                .build();
        streamSource.sinkTo(parquetSink2);


        env.execute();
    }
}
